package org.example.demo.broker;

import org.example.demo.domain.Message;
import org.example.demo.domain.MessageType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;


@Component
public class RabbitBrokerImpl implements RabbitBroker {
    private static final Logger LOGGER = LoggerFactory.getLogger(RabbitBrokerImpl.class);

    @Autowired
    private RabbitTemplateContainer container;

    @Override
    public void send(Message message) {
        if (message.getMessageType().equals(MessageType.RAPID)) {
            LOGGER.info("rapid send");
        } else if (message.getMessageType().equals(MessageType.CONFIRM)) {
            LOGGER.info("confirm send");
        } else {
            LOGGER.info("reliant send");
        }
        this.asyncSend(message);
    }

    private void asyncSend(Message message) {
        BaseAsyncQueue.submit(() -> {
            CorrelationData data = new CorrelationData(String.format("%s#%s#%s", message.getMessageId(), System.currentTimeMillis(), message.getMessageType()));
            String exchange = message.getExchange();
            String routingKey = message.getRoutingKey();
            RabbitTemplate template = this.container.getTemplate(message);
            template.convertAndSend(exchange, routingKey, message, data);
            LOGGER.info("send message to rabbitmq, messageId: {}", message.getMessageId());
        });
    }

    @Override
    public void sendMessages() {
        List<Message> messages = MessageHolder.getAndClear();
        messages.forEach(message -> {
            MessageHolderAsyncQueue.submit(() -> {
                CorrelationData data = new CorrelationData(String.format("%s#%s#%s", message.getMessageId(), System.currentTimeMillis(), message.getMessageType()));
                String exchange = message.getExchange();
                String routingKey = message.getRoutingKey();
                RabbitTemplate template = this.container.getTemplate(message);
                template.convertAndSend(exchange, routingKey, message, data);
                LOGGER.info("send message to rabbitmq, messageId: {}", message.getMessageId());
            });
        });
    }
}
